package cn.dmp.service

import java.util.Properties

import cn.dmp.util.{AppParams, Jpools}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}
import redis.clients.jedis.Jedis

/**
  * 3.2.1 地域分布
  */
object AreaDistributionV2 {


  /**
    * 地域分布     方法一：使用spark-sql 的 sqlContext
    *
    * @param parquet
    * @param sqlContext
    */
  def getAeraDistribution(parquet: DataFrame, sqlContext: SQLContext,broadInfo: Broadcast[Map[String, String]] ): Unit = {

    //3.2.3 媒体分析   appname: String, 应用名称
   /* etlDF.registerTempTable("t_appname_info")
    sqlContext.sql(
      """
        |select appname,count(1) totalRequest,sum(requestNum) requestNum,sum(validNum) validNum, sum(adRequestNum) adRequestNum,sum(participationNum) participationNum,
        |        sum(successfulsBidNum) successfulsBidNum,sum(showNum) showNum,sum(clickNum) clickNum,sum(dspConsume) dspConsume,sum(dspCost) dspCost
        |        from t_appname_info group by appname order by appname
      """.stripMargin).toDF.write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.appname, AppParams.prop)
    */
    // val parquetNew: DataFrame = parquet.filter("appid!=null or appid!='' or appname!=null or appname!=''")   //DataFrame DSL 语法

    //方式二：
    parquet.registerTempTable("t_log")
    // requestmode	processnode	iseffective	isbilling	isbid	iswin	adorderid

    /*sqlContext.udf.register("getAppName",(appname:String,appid:String)=>{
      val brocastMsg: Map[String, String] = broadInfo.value
      var appNameNew=if("0".equals(appname)||"".equals(appname)) brocastMsg.getOrElse(appid,appid) else appname
      appNameNew
    }) */

    sqlContext.udf.register("getAppNameRedis",(appname:String,appid:String)=>{
      val jedis: Jedis = Jpools.getJedis()
      var appNameNew=if("0".equals(appname)||"".equals(appname)) {
        val appTemp: String = jedis.hget("dict",appid)
        if(null==appTemp ||"".equals(appTemp)) appid else appTemp
      } else appname

      jedis.close()
      appNameNew
    })

    sqlContext.sql(
      """
        |select
        |getAppNameRedis(appname,appid) appname,
        |sum(1) totalRequest,
        |sum(case when requestmode=1 and processnode>=1 then 1 else 0 end) requestNum,
        |sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) validNum,
        |sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adRequestNum,
        |
        |sum(case when iseffective=1 and isbilling=1  and isbid=1 and adorderid!=0 then 1 else 0 end) participationNum,
        |sum(case when iseffective=1 and isbilling=1  and iswin=1 then 1 else 0 end) successfulsBidNum,
        |
        |sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) showNum,
        |sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) clickNum,
        |
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) dspConsume,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) dspCost
        |
        |from t_log
        |where appid !="" or appname !=""
        |group by getAppNameRedis(appname,appid)
      """.stripMargin).write.mode(SaveMode.Overwrite).jdbc(AppParams.url, AppParams.appnameV2, AppParams.prop)

  }


}
